package net.bwie.dws;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
//dws_refund_status（退款状态表）

public class Dws_refund_status {
    public static void main(String[] args) {

        // 1.创建环境
        TableEnvironment tableEnv = getTableEnv();

        // 2.创建输入表
        createInputTable(tableEnv);

        // 3.创建输出表
        createoutputTable(tableEnv);

        // 4.处理数据
        Table resultTable = handle(tableEnv);

//        tableEnv.createTemporaryView("dws_refund_status", resultTable);
//        tableEnv.executeSql("SELECT * FROM dws_refund_status").print();

        // 5.保存数据
        saveToSink(tableEnv, resultTable);
    }

    private static TableEnvironment getTableEnv() {
        EnvironmentSettings build = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnvironment = TableEnvironment.create(build);

        Configuration configuration = tableEnvironment.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("execution.checkpointing.interval", "5 s");

        return tableEnvironment;
    }

    private static void createInputTable(TableEnvironment tableEnv) {

        tableEnv.executeSql("CREATE TABLE dwd_retreat_behavior (\n" +
                //订单id
                "order_id STRING ,\n" +
                //  商品ID
                "product_id STRING ,\n" +
                // 订单支付时间
                "pay_time STRING ,\n" +

                // 退款单号
                "refund_id STRING ,\n" +
                // 退款申请时间
                "apply_time TIMESTAMP(3) ,\n" +
                //  退款场景:“退货退款”“已收货仅退款”“未收货退款”“未发货仅退款”
                "refund_scene STRING ,\n" +
                // 实际退款金额
                "refund_amount DOUBLE ,\n" +
                // 退款状态
                "refund_status STRING,\n"+

                //消费者行为id
                "behavior_id STRING ,\n" +
                //消费者id
                "user_id STRING ,\n" +
                //行为类型
                "behavior_type STRING, \n" +

                " time_ltz AS TO_TIMESTAMP_LTZ(UNIX_TIMESTAMP(pay_time),0),\n" +
                "WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' DAY\n" +
                ")WITH(\n" +
                "'connector' = 'kafka',\n" +
                "'topic' = 'dwd_retreat_behavior',\n" +
                "'properties.bootstrap.servers' = 'node101:9092',\n" +
                "'properties.group.id' = 'dwd_retreat_behavior-id08',\n" +
                "'scan.startup.mode' = 'earliest-offset',\n" +
                "'format' = 'json'\n" +
                ")"
        );

        //        TableResult tableResult4 = tableEnv.executeSql("SELECT * FROM dwd_retreat_behavior");
//        tableResult4.print();

    }

    private static void createoutputTable(TableEnvironment tableEnv) {
        tableEnv.executeSql("CREATE TABLE dws_refund_status (\n" +
                // 窗口开始时间
                "window_start TIMESTAMP(3),\n" +
                // 窗口结束时间
                "window_end TIMESTAMP(3),\n" +

                //（退款场景:“退货退款”“已收货仅退款”“未收货退款”“未发货仅退款”）
                "refund_scene STRING ,\n" +

                //退款人数
                "refunds_num INT,\n"+
                //  （退款子订单数）
                "refunds_zi_orders_num INT,\n"+
                //  （总订单数）
//                "total_order_num INT,\n"+
                //  （退款金额）
                "refund_amount DOUBLE\n"+
                //  （总金额）
//                "total_amount INT\n"+

                ")WITH(\n" +
                " 'connector' = 'doris',\n" +
                "'fenodes' = 'node102:8030',\n" +
                "'table.identifier' = 'transactions_dws.dws_refund_status',\n" +
                "'username' = 'root',\n" +
                "'password' = '123456',\n" +
                "'sink.label-prefix' = 'dws_refund_status_id10')\n"
        );
    }

    private static Table handle(TableEnvironment tableEnv) {

        Table tableResult =        tableEnv.sqlQuery(" SELECT\n" +
                " window_start ,\n" +
                " window_end,\n" +
                // 退款场景
                " refund_scene,\n" +
                " CAST(COUNT(user_id) as INT) AS refunds_num, -- 退款人数\n" +

                " sum(if(refund_id IS NOT NULL,1,0)) AS refunds_zi_orders_num, -- 退款子订单数\n" +

                " round(SUM(refund_amount),2) AS refund_amount -- 退款总金额\n" +
                " FROM TABLE(\n" +
                " -- 添加 1天 的 滚动窗口\n" +
                "  TUMBLE(TABLE dwd_retreat_behavior, DESCRIPTOR(time_ltz), INTERVAL '1' day)\n" +
                "                        )\n" +
//                "-- 筛选有效退款场景（四类场景）\n" +
                "    WHERE  refund_scene IN ('退货退款', '已收货仅退款', '未收货退款', '未发货仅退款')\n" +
                "    GROUP BY\n" +
                "        window_start ,\n" +
                "        window_end,\n" +
                "         refund_scene\n" +
                "         ");
        return tableResult;
    }

    private static void saveToSink(TableEnvironment tableEnv, Table resultTable) {
        tableEnv.createTemporaryView("resultTable", resultTable);
        tableEnv.executeSql("INSERT INTO dws_refund_status SELECT * FROM resultTable");
    }


}
